package ru.dadle8.serializers;

import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;
import ru.dadle8.model.Customer;

import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.util.Map;

public class CustomerDeserializer implements Deserializer<Customer> {

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        Deserializer.super.configure(configs, isKey);
    }

    @Override
    public Customer deserialize(String s, byte[] bytes) {
        try {
            if (bytes == null) {
                System.out.println("Null received at deserializing");
                return null;
            }

            try (ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
                 ObjectInputStream ois = new ObjectInputStream(bis)) {
                return (Customer) ois.readObject();
            }
//            ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
//            int customerId = byteBuffer.getInt();
//            int stringSize = byteBuffer.getInt();
//
//            String customerName = new String(Arrays.copyOfRange(bytes, Integer.BYTES + Integer.BYTES, Integer.BYTES + Integer.BYTES + stringSize), StandardCharsets.UTF_8);
//            return new Customer(customerId, customerName);
        } catch (Exception e) {
            throw new SerializationException("Error when deserializing byte[] to Customer");
        }
    }

    @Override
    public void close() {
        Deserializer.super.close();
    }
}
